package com.atguigu.userprofile.app;

import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.bean.TaskInfo;
import com.atguigu.userprofile.bean.TaskTagRule;
import com.atguigu.userprofile.constants.ConstCodes;
import com.atguigu.userprofile.dao.TagInfoDAO;
import com.atguigu.userprofile.dao.TaskInfoDAO;
import com.atguigu.userprofile.dao.TaskTagRuleDAO;
import com.atguigu.userprofile.utils.MyPropertiesUtil;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.SparkSession;


import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;

/**
 * ClassName: TaskSQLApp
 * Package: com.atguigu.userprofile.app
 * Description:
 *
 * @Author ChenJun
 * @Create 2023/3/29 11:11
 * @Version 1.0
 */
public class TaskSQLApp {


    public static void main(String[] args) {

        // 获取spark运行环境
        //TODO 注意！！！！ 打包上传的时候要注释掉setMaster
        SparkConf sparkConf = new SparkConf().setAppName("TaskSQLApp");//.setMaster("local[*]");
//        SparkConf sparkConf = new SparkConf().setAppName("TaskSQLApp").setMaster("local[*]");

        SparkSession sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate();

        //0. 获取配置文件中的信息
        Properties properties = MyPropertiesUtil.load("config.properties");
        String hdfsPath = properties.getProperty("hdfs-store.path");
        String dwDbName = properties.getProperty("data-warehouse.dbname");
        String upDbName = properties.getProperty("user-profile.dbname");


        // 1.要获取外部 传参 task_id 、busiDate
          String taskId = args[0];
          String busiDate = args[1];
        //System.out.println(taskId);

        // 2.根据 taskId 查询 Mysql中 查询tag_info、task_info、task_tag_rule这三张表的数据
        // 2.1获取tagInfo
        TagInfo tagInfo = TagInfoDAO.getTagInfo(taskId);
//         System.out.println("tagInfo--->"+ tagInfo);

        // 2.2获取taskInfo
        TaskInfo taskInfo = TaskInfoDAO.getTaskInfo(taskId);
        // System.out.println(taskInfo);

        // 2.3 获取taskTagRules
        List<TaskTagRule> taskTagRuleList = TaskTagRuleDAO.getTaskTagRule(taskId);
        // System.out.println(taskTagRulesList);

        // 3.动态拼接建表语句
        // 3.1 获取表名 并转换成小写
        String tableName = tagInfo.getTagCode().toLowerCase();

        // 3.2 获取值的类型
        String tagValueType = tagInfo.getTagValueType();

        // 判断设置值的类型
        String valueTypeSQL = null;
        if (ConstCodes.TAG_VALUE_TYPE_LONG.equals(tagValueType)){
            valueTypeSQL = "Long";
        }else if (ConstCodes.TAG_VALUE_TYPE_DECIMAL.equals(tagValueType)){
            valueTypeSQL = "decimal";
        }else if (ConstCodes.TAG_VALUE_TYPE_STRING.equals(tagValueType)){
            valueTypeSQL = "string";
        }else if (ConstCodes.TAG_VALUE_TYPE_DATE.equals(tagValueType)){
            valueTypeSQL = "string";
        }

        //3.3 获取表名注释
        String commentSQL = tagInfo.getTagName();

        // 建表语句
        String createTableSQL = "create table if not exists "+upDbName+"."+tableName+" \n" +
                "  (uid String,tag_value "+valueTypeSQL+")\n" +
                "  partitioned by (dt String)\n" +
                "  comment '"+commentSQL+"' \n" +
                "  ROW FORMAT DELIMITED FIELDS TERMINATED BY '\\t' \n" +
                "  location '"+hdfsPath+"/"+upDbName+"/"+tableName+"'";

//        System.out.println(createTableSQL);


        // 4.动态拼接 查询&插入语句
        // 4.1 拼接查询语句
        /**
         * select
         *  uid,
         *  case query_value when 'F' then '男'  when 'M' then '女'  when 'U' then '未知'  end as query_value
         *  from
         *  (
         *    select
         *     id as uid,
         *     if(gender<>"",gender,"U")  as query_value
         *    from
         *     dim_user_zip where dt='9999-12-31'
         *   )ti
         */
        //将集合中的元素转换为 when...then..格式
        List<String> whenThenSQlList = taskTagRuleList.stream().map(taskTagRule -> "when '" + taskTagRule.getQueryValue() + "' then '" + taskTagRule.getSubTagValue() + "'").collect(Collectors.toList());
        //System.out.println(whenThenSQLList);

        //将集合中每个元素拼起来拼成一个字符串
        // whenThen 语句
        // 方式一：
        String whenThenSQL = String.join(" ", whenThenSQlList);
        //System.out.println(whenThenSQL);

        //方式二：
        //String whenThenSQL = StringUtils.join(whenThenSQlList," ");
//        System.out.println(whenThenSQL);

        // case... when.. then .. 语句
        String caseWhenSQL = "case query_value "+whenThenSQL+" end as query_value";

        //MYSQL  表task_info中的 task_sql语句
        String subSQLString = taskInfo.getTaskSql();


        //考虑有的标签没有便签值  需要对集合中的元素个数进行判断 是否走caseWhenThen
        String querySQL="";

        if (taskTagRuleList.size()>0) {
            querySQL = " select\n" +
                    "           uid,\n" +
                    "           " + caseWhenSQL + "\n" +
                    "           from\n" +
                    "           ("+subSQLString+")ti";
        }else {
            querySQL=" select\n" +
                    "           uid,\n" +
                    "           queury_value \n" +
                    "           from\n" +
                    "           ("+subSQLString+")ti";
        }
        //System.out.println(querySQL);

        // 数据插入语句
        /**
         *  insert overwrite table user_profile0409.tg_person_base_gender partition(dt='2020-06-14') select
         *  uid,
         *  case query_value when 'F' then '女'  when 'M' then '男'  when 'U' then '未知'   end as query_value
         *  from
         *  (select id as uid,if(gender<>"",gender,"U") as query_value from dim_user_zip where dt='9999-12-31')ti
         */

        // 插入语句
        String insertSQL = " insert overwrite table "+upDbName+"."+tableName+" partition(dt='"+busiDate+"') " + querySQL + "";
        System.out.println(insertSQL);

        // 5.执行sql
        //指定在哪个库里面运行
        sparkSession.sql("use "+dwDbName+"");
        sparkSession.sql(createTableSQL);
        sparkSession.sql(insertSQL);


    }
}
